package com.rrd.dw.mr.gzip.kafka;

import java.io.IOException;
import java.util.Date;
import java.util.List;
import java.util.regex.Pattern;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import com.rrd.dw.utils.*;
import com.ucredit.babel.common.kafka.CrawlResult;
import com.ucredit.babel.common.model.Job;
import com.ucredit.babel.common.model.call.CallDetail;
import com.ucredit.babel.common.model.call.CallRecord;
import com.ucredit.babel.common.model.call.CallUserInfo;
import com.ucredit.babel.common.model.call.NetFlow;
import com.ucredit.babel.common.model.call.SmsBill;

@Deprecated
public class CallRecordParse {
	public static final String CRT001 =  "\001";
	public static final Pattern CRT001_PATTERN =  Pattern.compile("\001");
	private static Text val = new Text();
	private Text key= new Text();
	public CallRecordParse(){

	}
	
	/**
	 * 构建运营商用户数据
	 * @param callRecord
	 * @throws InterruptedException 
	 * @throws IOException 
	 */
	 public  void parseCallRecordUser(Context context,CallRecord callRecord,String successful) throws IOException, InterruptedException{
	    	StringBuilder sb = new StringBuilder(200);
	    	Job job = callRecord.getJob();
	    	CrawlResult result=job.getResult();
	    	CallUserInfo user=callRecord.getUserInfo();
	    	if(user==null){
	    		user = new CallUserInfo();
	    	}
	    	String s="";
	    	//爬取时间
	    	sb.append(DateUtil.formateDateStr(new Date(callRecord.getTimestamp()))).append(CRT001);
	    	//身份证
	    	sb.append(callRecord.getIdNumber()==null?"":callRecord.getIdNumber()).append(CRT001);
	    	//姓名
	    	sb.append(user.getRealName()==null?"":user.getRealName()).append(CRT001);
	    	//入网时间
	    	sb.append(user.getInNetDate()==null?"":DateUtil.formateDateStr(user.getInNetDate())).append(CRT001);
	    	//用户级别
	    	sb.append(user.getLevel()==null?"":user.getLevel()).append(CRT001);
	    	//邮箱
	    	sb.append(user.getEmail()==null?"":user.getEmail()).append(CRT001);
	    	//住址
	    	sb.append(user.getAddress()==null?"":user.getAddress()).append(CRT001);
	    	//邮编
	    	sb.append(user.getZipCode()==null?"":user.getZipCode()).append(CRT001);
	    	//手机号
	    	sb.append(user.getMobile()==null?"":user.getMobile()).append(CRT001);
	    	//星级得分
	    	sb.append(user.getStarScore()==null?"":user.getStarScore()).append(CRT001);
	    	//客户星级
	    	sb.append(user.getStarLevel()==null?"":user.getStarLevel()).append(CRT001);
	    	//用户状态
	    	sb.append(user.getStatus()==null?"":user.getStatus()).append(CRT001);
	    	//性别
	    	sb.append(user.getGender()==null?"":user.getGender()).append(CRT001);
	    	//证件类型
	    	sb.append(user.getCertType()==null?"":user.getCertType()).append(CRT001);
	    	//证件号
	    	sb.append(user.getCertNumber()==null?"":user.getCertNumber()).append(CRT001);
	    	//手机运营商
	    	if(user.getCarrierCode()==null){
	    		user.setCarrierCode("");
	    	}else if(user.getCarrierCode().startsWith("CHINAMOBILE")){
	    		user.setCarrierCode("10086");
	    	}else if(user.getCarrierCode().startsWith("CHINANET")){
	    		user.setCarrierCode("10000");
	    	}else if(user.getCarrierCode().startsWith("CHINAUNICOM")){
	    		user.setCarrierCode("10010");
	    	}
	    	sb.append(user.getCarrierCode()).append(CRT001);
	    	//电话所属地
	    	sb.append(user.getProvince()==null?"":user.getProvince()).append(CRT001);
	    	//余额
	    	sb.append(user.getBalance()==null?"":user.getBalance()).append(CRT001);
	    	//数据来源
	    	sb.append(user.getDataSource()==null?"":user.getDataSource()).append(CRT001);
	    	//jobid
	    	sb.append(job.getJobID()==null?"":job.getJobID()).append(CRT001);
	    	//是否爬取完成
	    	sb.append(job.isFinished()!=null && job.isFinished()?"1":"0").append(CRT001);
	    	//是否有请求 
	    	sb.append(job.isRequested()!=null && job.isRequested()?"1":"0").append(CRT001);
	    	//是否爬取成功 
//	    	sb.append(job.isSuccessful()!=null && job.isSuccessful()?"1":"0").append(CRT001);
	    	sb.append(successful).append(CRT001);
	    	//爬取开始时间
	    	sb.append(job.getStartTime()==null?"":DateUtil.formateDateStr(job.getStartTime())).append(CRT001);
	    	//爬取结束时间
	    	sb.append(job.getEndTime()==null?"":job.getEndTime()).append(CRT001);
	    	//文件附件
	    	sb.append(job.getDataFile()==null?"":job.getDataFile()).append(CRT001);
	    	//数据爬取业务系统
	    	sb.append(job.getSystemID()==null?"":job.getSystemID()).append(CRT001);
	    	//手机号码
	    	sb.append(job.getQuery()==null?"":job.getQuery()).append(CRT001);
	    	//重试次数
	    	sb.append(job.getTryTimes()==null?"":job.getTryTimes()).append(CRT001);
	    	//爬取结果类型 SUCCESSFUL
//	    	sb.append(result==null?"":result.getType()).append(CRT001);
	    	sb.append("SUCCESSFUL").append(CRT001);
	    	//爬取结果消息
	    	sb.append(result==null?"":result.getMessage()).append(CRT001);
	    	//消息code
	    	sb.append(result==null?"":result.getCode()).append(CRT001);
	    	//时间戳
	    	sb.append(callRecord.getTimestamp()==null?"":DateUtil.formateDateStr(new Date(callRecord.getTimestamp())));
	    	key.set("call_user");
	    	val.set(sb.toString().replace("\n", " "));
	    	sb.delete(0, sb.length());
	    	
	    	context.write(key, val);
	    	
	    }
	 
	 /**
	  * 运营商短信详单数据
	  * @param callRecord
	 * @throws InterruptedException 
	 * @throws IOException 
	  */
	 public  void parseCallRecordSms(Context context,CallRecord callRecord) throws IOException, InterruptedException{
		StringBuilder sb = new StringBuilder(200);
	    Job job = callRecord.getJob();
	    List<SmsBill> list=callRecord.getSmsBills();
	    int i=0;
	    for(SmsBill sms:list){
	    	i++;
	    	//jobid
	    	sb.append(job.getJobID()).append(CRT001);
	    	sb.append(DateUtil.formateDateStr(new Date(callRecord.getTimestamp()))).append(CRT001);
	    	//身份证 getIdNumber
	    	sb.append(callRecord.getIdNumber()==null?"":callRecord.getIdNumber()).append(CRT001);
//	    	sb.append(user.getCertNumber()==null?"":user.getCertNumber()).append(CRT001);
	    	//短信开始时间(时间戳)
	    	sb.append(sms.getSmsStartTime()==null?"":DateUtil.formateDateStr(sms.getSmsStartTime())).append(CRT001);
	    	//地区
	    	sb.append(sms.getSmsArea()==null?"":sms.getSmsArea()).append(CRT001);
	    	//对方手机号
	    	sb.append(sms.getSmsToMobile()==null?"":sms.getSmsToMobile()).append(CRT001);
	    	//方式 1：发送 其他：2
	    	sb.append(sms.getSmsMode()==null?"":sms.getSmsMode()).append(CRT001);
	    	//短信类型 1：短信 2：彩信
	    	sb.append(sms.getSmsInfoType()==null?"":sms.getSmsInfoType()).append(CRT001);
	    	//业务名称
	    	sb.append(sms.getSmsBusinessName()==null?"":sms.getSmsBusinessName()).append(CRT001);
	    	//总费用(单位：分)
	    	sb.append(sms.getSmsTotalFee()==null?"":sms.getSmsTotalFee()).append(CRT001);
	    	//本人手机号
	    	sb.append(sms.getSmsMobile()==null?"":sms.getSmsMobile()).append(CRT001);
	    	//job_id下的序号
	    	sb.append(String.valueOf(i));
	    	key.set("call_sms");
	    	val.set(sb.toString());
	    	sb.delete(0, sb.length());
	    	
	    	context.write(key, val);
	    	
	    }
	 }
	 /**
	  * 运营商流量详单数据
	  * @param callRecord
	 * @throws InterruptedException 
	 * @throws IOException 
	  */
	 public  void parseCallRecordNF(Context context,CallRecord callRecord) throws IOException, InterruptedException{
		StringBuilder sb = new StringBuilder(200);
	    Job job = callRecord.getJob();
	    List<NetFlow> list=callRecord.getNetFLows();
	    int i =0;
	    for(NetFlow nf:list){
	    	i++;
	    	//jobid
	    	sb.append(job.getJobID()).append(CRT001);
	    	//爬取时间
	    	sb.append(callRecord.getTimestamp()==null?"":DateUtil.formateDateStr(new Date(callRecord.getTimestamp()))).append(CRT001);
	    	//身份证
//	    	sb.append(user.getCertNumber()==null?"":user.getCertNumber()).append(CRT001);
	    	sb.append(callRecord.getIdNumber()==null?"":callRecord.getIdNumber()).append(CRT001);
	    	//开始时间(时间戳)
	    	sb.append(nf.getNfStartTime()==null?"":DateUtil.formateDateStr(nf.getNfStartTime())).append(CRT001);
	    	//地区
	    	sb.append(nf.getNfArea()==null?"":nf.getNfArea()).append(CRT001);
	    	//网络类型
	    	sb.append(nf.getNfNetType()==null?"":nf.getNfNetType()).append(CRT001);
	    	//上网时长(单位：秒)
	    	sb.append(nf.getNfDuration()==null?"":nf.getNfDuration()).append(CRT001);
	    	//流量类型 套餐内,套餐外
	    	sb.append(nf.getNfFlowType()==null?"":nf.getNfFlowType()).append(CRT001);
	    	//总流量(单位：KB) 
	    	sb.append(nf.getNfTotalFlows()==null?"":nf.getNfTotalFlows()).append(CRT001);
	    	//总费用(单位：分)
	    	sb.append(nf.getNfTotalFee()==null?"":nf.getNfTotalFee()).append(CRT001);
	    	//手机号
	    	sb.append(nf.getNfMobile()==null?"":nf.getNfMobile()).append(CRT001);
	    	//数据归属
	    	sb.append(nf.getNfSource()==null?"":nf.getNfSource()).append(CRT001);
	    	//job_id下的序号
	    	sb.append(String.valueOf(i));
	    	key.set("call_netflows");
	    	val.set(sb.toString());
	    	sb.delete(0, sb.length());
	    	
	    	context.write(key, val);
	    }
	 }
	 
	 /**
	  * 运营商通话详单数据
	  * @param callRecord
	 * @throws InterruptedException 
	 * @throws IOException 
	  */
	 public  void parseCallRecordDetail(Context context,CallRecord callRecord) throws IOException, InterruptedException{
		StringBuilder sb = new StringBuilder(200);
	    Job job = callRecord.getJob();
	    List<CallDetail> list=callRecord.getDetails();
	    int i=0;
	    for(CallDetail c:list){
	    	i++;
	    	//jobid
	    	sb.append(job.getJobID()==null?"":job.getJobID()).append(CRT001);
	    	//爬取时间
	    	sb.append(callRecord.getTimestamp()==null?"":DateUtil.formateDateStr(new Date(callRecord.getTimestamp()))).append(CRT001);
	    	//身份证
//	    	sb.append(user.getCertNumber()==null?"":user.getCertNumber()).append(CRT001);
	    	sb.append(callRecord.getIdNumber()==null?"":callRecord.getIdNumber()).append(CRT001);
	    	//通话开始时间
	    	sb.append(c.getStartTime()==null?"":DateUtil.formateDateStr(c.getStartTime())).append(CRT001);
	    	//通话发起所在地
	    	sb.append(c.getArea()==null?"":c.getArea()).append(CRT001);
	    	//	呼叫类型 1 主叫 2 被叫 
	    	sb.append(c.getMode()==null?"":c.getMode()).append(CRT001);
	    	//对方号码
	    	sb.append(c.getToMobile()==null?"":c.getToMobile()).append(CRT001);
	    	//自己手机号
	    	sb.append(c.getMobile()==null?"":c.getMobile()).append(CRT001);
	    	//通话时长（int 单位：秒） 
	    	sb.append(c.getDuration()==null?"":c.getDuration()).append(CRT001);
	    	//通话类型 本地通话、国内漫游、国际漫游
	    	sb.append(c.getType()==null?"":c.getType()).append(CRT001);
	    	//费用小计（单位：分）
	    	sb.append(c.getTotalFee()==null?"":c.getTotalFee()).append(CRT001);
	    	//手机运营商代码
	    	sb.append(c.getCarrierCode()==null?"":c.getCarrierCode()).append(CRT001);
	    	//普通话费(单位：分)
	    	sb.append(c.getNormalFee()==null?"":c.getNormalFee()).append(CRT001);
	    	//长途话费(单位：分)
	    	sb.append(c.getRoamingFee()==null?"":c.getRoamingFee()).append(CRT001);
	    	//job_id下的序号
	    	sb.append(String.valueOf(i));
	    	key.set("call_details");
	    	val.set(sb.toString());
	    	sb.delete(0, sb.length());
	    	
	    	context.write(key, val);
	    }
	 }
	 
	 public static void main(String[] args) {
		
		 StringBuilder sb = new StringBuilder(200);
		 sb.append("abcddfffffffffffffffffff");
		 sb.delete(0, sb.length());
		 System.out.println(sb.toString());
	}
}
